Corporate Events Processor

Overview

The add_corporate_events.py script is the final enrichment layer in the EDL pipeline. It consolidates multiple data sources to add event markers, corporate action calendars, insider trading alerts, surveillance status, and news feeds to the master analysis file.

Purpose

This comprehensive script enriches stock data with:

Surveillance markers: LTASM, STASM, GSM status
Corporate actions: Upcoming dividends, bonuses, splits, rights issues, results
Circuit limit revisions: Positive and negative changes
Block/bulk deals: Recent large transactions
Insider trading alerts: Regulatory disclosures
Recent announcements: Top 5 regulatory filings
News feed: Market sentiment and media coverage

Input Files Required

all_stocks_fundamental_analysis.json

JSON

required

Master analysis file. This is both input and output.

upcoming_corporate_actions.json

JSON

Calendar of upcoming corporate actions with ex-dates.

nse_asm_list.json

JSON

NSE Additional Surveillance Measure (ASM) list with LTASM/STASM stages.

nse_gsm_list.json

JSON

NSE Graded Surveillance Measure (GSM) list.

bulk_block_deals.json

JSON

Recent bulk and block deal transactions.

incremental_price_bands.json

JSON

Circuit limit revision history.

company_filings/

Output Produced

all_stocks_fundamental_analysis.json

JSON

Updates the master analysis file with event markers, announcements, and news feeds.

Processing Logic

1. Surveillance Status Mapping

Marks stocks under regulatory surveillance:

def add_event(sym, event_str):
    if sym not in refined_map:
        refined_map[sym] = []
    if event_str not in refined_map[sym]:
        refined_map[sym].append(event_str)

# Process ASM List
if os.path.exists(asm_file):
    with open(asm_file, "r") as f:
        asm_data = json.load(f)
        for item in asm_data:
            sym = item.get("Symbol")
            stage = item.get("Stage", "")
            if sym:
                if "LTASM" in stage:
                    add_event(sym, "★: LTASM")
                elif "STASM" in stage:
                    add_event(sym, "★: STASM")

2. Corporate Actions Processing

Maps upcoming events with date-based filtering:

if os.path.exists(upcoming_file):
    with open(upcoming_file, "r") as f:
        upcoming_data = json.load(f)
        today = datetime.now()
        action_limit = today + timedelta(days=30)
        results_limit = today + timedelta(days=14)
        
        for event in upcoming_data:
            sym = event.get("Symbol")
            etype = event.get("Type", "")
            edate_str = event.get("ExDate")
            
            if not sym or not edate_str: continue
            
            edate = datetime.strptime(edate_str, "%Y-%m-%d")
            if today.date() <= edate.date() <= action_limit.date():
                d_str = edate.strftime("%d-%b")
                if "QUARTERLY" in etype and edate.date() <= results_limit.date():
                    add_event(sym, f"⏰: Results ({d_str})")
                elif "DIVIDEND" in etype:
                    add_event(sym, f"💸: Dividend ({d_str})")
                elif "BONUS" in etype:
                    add_event(sym, f"🎁: Bonus ({d_str})")
                elif "SPLIT" in etype:
                    add_event(sym, f"✂️: Split ({d_str})")
                elif "RIGHTS" in etype:
                    add_event(sym, f"📈: Rights ({d_str})")

3. Circuit Limit Revisions

Detects changes in price bands:

if os.path.exists(circuit_revision_file):
    with open(circuit_revision_file, "r") as f:
        rev_data = json.load(f)
        for item in rev_data:
            sym = item.get("Symbol")
            f_band = item.get("From")
            t_band = item.get("To")
            if sym and f_band and t_band:
                try:
                    if float(t_band) < float(f_band):
                        add_event(sym, "#: -ve Circuit Limit Revision")
                    elif float(t_band) > float(f_band):
                        add_event(sym, "#: +ve Circuit Limit Revision")
                except:
                    pass

4. Block and Bulk Deals

Tracks recent large transactions (last 7 days):

if os.path.exists(deals_file):
    with open(deals_file, "r") as f:
        deals_data = json.load(f)
        today = datetime.now()
        recent_limit = today - timedelta(days=7)
        
        for deal in deals_data:
            sym = deal.get("sym")
            dtype = deal.get("deal", "")
            d_date_str = deal.get("date", "").split(" ")[0]
            
            if sym and d_date_str:
                d_date = datetime.strptime(d_date_str, "%Y-%m-%d")
                if d_date >= recent_limit:
                    if "BLOCK" in dtype or "BULK" in dtype:
                        add_event(sym, "📦: Block Deal")

5. Insider Trading Detection

Scans regulatory filings for insider trading disclosures:

for item in items:
    desc = (item.get("descriptor") or "").lower()
    caption = (item.get("caption") or "").lower()
    cat = (item.get("cat") or "").lower()
    body = (item.get("news_body") or "").lower()
    n_date_str = item.get("news_date", "").split(" ")[0]
    
    if n_date_str:
        n_date = datetime.strptime(n_date_str, "%Y-%m-%d")
        if n_date >= recent_limit:
            is_insider = False
            full_text = f"{desc} {caption} {cat} {body}"
            
            trade_keywords = ["regulation 7(2)", "reg 7(2)", "inter-se transfer", "form c", "continual disclosure"]
            if any(k in full_text for k in trade_keywords):
                is_insider = True
            elif ("insider trading" in full_text or "sebi (pit)" in full_text or "sebi pit" in full_text):
                if "trading window" not in full_text and "closure" not in full_text:
                    is_insider = True
            
            if is_insider:
                add_event(sym, "🔑: Insider Trading")
                break

6. Recent Announcements Aggregation

Captures top 5 regulatory filings:

news_map[sym] = []
for item in items[:5]:
    headline = (item.get("caption") or item.get("descriptor") or item.get("news_body") or "N/A")
    headline = " ".join(headline.split())
    news_map[sym].append({
        "Date": item.get("news_date", "N/A"),
        "Headline": headline,
        "URL": item.get("file_url") or "N/A"
    })

7. Results Recently Out Marker

Identifies stocks with recent results announcements:

if os.path.exists(ann_file):
    with open(ann_file, "r") as f:
        ann_data = json.load(f)
        today = datetime.now()
        marker_limit = today - timedelta(days=7)
        
        for ann in ann_data:
            sym = ann.get("Symbol")
            event_text = (ann.get("Event") or "")
            etype = ann.get("Type", "")
            date_str = ann.get("Date", "")
            
            if sym and date_str:
                a_date = datetime.strptime(date_str.split(" ")[0], "%Y-%m-%d")
                if a_date >= marker_limit:
                    if "results are out" in event_text.lower() or etype == "Results Update":
                        add_event(sym, "📊: Results Recently Out")

8. Market News Feed Integration

Adds sentiment-analyzed market news:

market_news_dir = os.path.join(BASE_DIR, "market_news")
news_feed_map = {}

if os.path.exists(market_news_dir):
    news_files = glob.glob(os.path.join(market_news_dir, "*_news.json"))
    for nf in news_files:
        with open(nf, "r") as f:
            n_data = json.load(f)
            sym = n_data.get("Symbol")
            news_list = n_data.get("News", [])
            
            if sym and news_list:
                formatted_news = []
                for item in news_list[:5]:
                    formatted_news.append({
                        "Title": item.get("Title"),
                        "Sentiment": item.get("Sentiment"),
                        "Date": item.get("PublishDate")
                    })
                news_feed_map[sym] = formatted_news

9. Master Data Update

Applies all enrichments to the master file:

for stock in master_data:
    sym = stock.get("Symbol")
    
    # Update Events
    events = refined_map.get(sym, [])
    stock["Event Markers"] = " | ".join(events) if events else "N/A"
    
    # Update Recent Announcements (Top 5 - Regulatory)
    stock["Recent Announcements"] = news_map.get(sym, [])[:5]
    
    # Update News Feed (Top 5 - Market/Media)
    stock["News Feed"] = news_feed_map.get(sym, [])

with open(master_file, "w") as f:
    json.dump(master_data, f, indent=4, ensure_ascii=False)

Fields Added/Modified

This script adds the following fields:

Event Markers

Event Markers: Pipe-delimited string of active event markers

Marker Types:

★: LTASM - Long Term Additional Surveillance Measure
★: STASM - Short Term Additional Surveillance Measure
⏰: Results (DD-MMM) - Upcoming quarterly results
💸: Dividend (DD-MMM) - Upcoming dividend ex-date
🎁: Bonus (DD-MMM) - Upcoming bonus issue
✂️: Split (DD-MMM) - Upcoming stock split
📈: Rights (DD-MMM) - Upcoming rights issue
#: -ve Circuit Limit Revision - Circuit limit reduced
#: +ve Circuit Limit Revision - Circuit limit increased
📦: Block Deal - Recent block/bulk deal (last 7 days)
🔑: Insider Trading - Recent insider trading disclosure (last 15 days)
📊: Results Recently Out - Results announced in last 7 days

Announcements and News

Recent Announcements: Array of top 5 regulatory filings with date, headline, and URL
News Feed: Array of top 5 market news items with title, sentiment, and date

Code Example

add_corporate_events.py

import json
import glob
from datetime import datetime, timedelta

def map_refined_events():
    BASE_DIR = os.path.dirname(os.path.abspath(__file__))
    master_file = os.path.join(BASE_DIR, "all_stocks_fundamental_analysis.json")
    
    with open(master_file, "r") as f:
        master_data = json.load(f)
    
    refined_map = {}
    
    def add_event(sym, event_str):
        if sym not in refined_map:
            refined_map[sym] = []
        if event_str not in refined_map[sym]:
            refined_map[sym].append(event_str)
    
    # Process surveillance
    with open(asm_file, "r") as f:
        asm_data = json.load(f)
        for item in asm_data:
            sym = item.get("Symbol")
            stage = item.get("Stage", "")
            if sym:
                if "LTASM" in stage:
                    add_event(sym, "★: LTASM")
    
    # Process corporate actions
    with open(upcoming_file, "r") as f:
        upcoming_data = json.load(f)
        for event in upcoming_data:
            sym = event.get("Symbol")
            etype = event.get("Type")
            edate = datetime.strptime(event.get("ExDate"), "%Y-%m-%d")
            if "DIVIDEND" in etype:
                add_event(sym, f"💸: Dividend ({edate.strftime('%d-%b')})")
    
    # Update master data
    for stock in master_data:
        sym = stock.get("Symbol")
        events = refined_map.get(sym, [])
        stock["Event Markers"] = " | ".join(events) if events else "N/A"
    
    with open(master_file, "w") as f:
        json.dump(master_data, f, indent=4, ensure_ascii=False)

Function Reference

`add_event(sym, event_str)`

Adds an event marker to a symbol’s event list (helper function within map_refined_events). Parameters:

sym: Stock symbol
event_str: Formatted event marker string

Returns: None (updates global refined_map dictionary)

`map_refined_events()`

Main orchestration function that processes all event sources and updates the master JSON. Returns: None (writes output to JSON file)

Processing Windows

Event Type	Lookback/Forward Window
Corporate Actions (General)	Next 30 days
Upcoming Results	Next 14 days
Block/Bulk Deals	Last 7 days
Insider Trading	Last 15 days
Results Recently Out	Last 7 days
Announcements	Top 5 (all time)
News Feed	Top 5 (all time)

Event Marker Examples

Example 1: Multiple Events

Symbol: RELIANCE
Event Markers: "💸: Dividend (15-Mar) | 📊: Results Recently Out | #: +ve Circuit Limit Revision"

Example 2: Surveillance

Symbol: EXAMPLE
Event Markers: "★: STASM | 📦: Block Deal"

Example 3: Corporate Actions Calendar

Symbol: INFY
Event Markers: "⏰: Results (20-Apr) | 💸: Dividend (22-Apr)"

Announcement Structure

"Recent Announcements": [
  {
    "Date": "2026-02-15 18:30:00",
    "Headline": "Outcome of Board Meeting - Approval of Financial Results",
    "URL": "https://www.nseindia.com/corporates/..."
  },
  ...
]

News Feed Structure

"News Feed": [
  {
    "Title": "Company announces major expansion plans",
    "Sentiment": "Positive",
    "Date": "2026-03-01T10:30:00Z"
  },
  ...
]

Performance Notes

Processing Time: ~2,000 stocks processed in 10-15 seconds
File I/O: Multiple file reads; could benefit from caching
Memory Usage: Moderate (all news/filings loaded into memory)
Sequential Processing: No parallelization

Dependencies

json: JSON file handling
os: File path operations
glob: File pattern matching for news and filings directories
datetime: Date filtering and formatting

Important Notes

Final Pipeline Stage: Must run last after all other processors
Graceful Degradation: Missing data sources result in “N/A” or empty arrays
Unicode Support: Uses ensure_ascii=False for proper emoji rendering
Date Formats: Expects YYYY-MM-DD for consistency across sources
Insider Trading Logic: Sophisticated keyword matching to reduce false positives
Deduplication: Prevents duplicate event markers and news items

Source File Location

add_corporate_events.py:1-264

Data Fetching Scripts

Processing Scripts

Standalone Scripts

Utilities

Overview

Purpose

Input Files Required

Output Produced

Processing Logic

1. Surveillance Status Mapping

2. Corporate Actions Processing

3. Circuit Limit Revisions

4. Block and Bulk Deals

5. Insider Trading Detection

6. Recent Announcements Aggregation

7. Results Recently Out Marker

8. Market News Feed Integration

9. Master Data Update

Fields Added/Modified

Event Markers

Marker Types:

Announcements and News

Code Example

Function Reference

`add_event(sym, event_str)`

`map_refined_events()`

Processing Windows

Event Marker Examples

Example 1: Multiple Events

Example 2: Surveillance

Example 3: Corporate Actions Calendar

Announcement Structure

News Feed Structure

Performance Notes

Dependencies

Important Notes

Source File Location

Data Fetching Scripts

Processing Scripts

Standalone Scripts

Utilities

​Overview

​Purpose

​Input Files Required

​Output Produced

​Processing Logic

​1. Surveillance Status Mapping

​2. Corporate Actions Processing

​3. Circuit Limit Revisions

​4. Block and Bulk Deals

​5. Insider Trading Detection

​6. Recent Announcements Aggregation

​7. Results Recently Out Marker

​8. Market News Feed Integration

​9. Master Data Update

​Fields Added/Modified

​Event Markers

​Marker Types:

​Announcements and News

​Code Example

​Function Reference

​add_event(sym, event_str)

​map_refined_events()

​Processing Windows

​Event Marker Examples

​Example 1: Multiple Events

​Example 2: Surveillance

​Example 3: Corporate Actions Calendar

​Announcement Structure

​News Feed Structure

​Performance Notes

​Dependencies

​Important Notes

​Source File Location

Overview

Purpose

Input Files Required

Output Produced

Processing Logic

1. Surveillance Status Mapping

2. Corporate Actions Processing

3. Circuit Limit Revisions

4. Block and Bulk Deals

5. Insider Trading Detection

6. Recent Announcements Aggregation

7. Results Recently Out Marker

8. Market News Feed Integration

9. Master Data Update

Fields Added/Modified

Event Markers

Marker Types:

Announcements and News

Code Example

Function Reference

`add_event(sym, event_str)`

`map_refined_events()`

Processing Windows

Event Marker Examples

Example 1: Multiple Events

Example 2: Surveillance

Example 3: Corporate Actions Calendar

Announcement Structure

News Feed Structure

Performance Notes

Dependencies

Important Notes

Source File Location